package mapReduce.demo05_shuffle.demo01_phoneTrafficPartByKey;

import mapReduce.writable.PhoneTrafficWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

public class PhoneTrafficPartitioner extends Partitioner<Text, PhoneTrafficWritable> {

    /**
     * 对于mapper端输出的key可以value 按照自己定义给的规则进行分区，每个分区就用索引表示即可
     * 需求：
     *      按照号段进行分区
     *          130-149 进入第 0 分区
     *          150-169 进入第 1 分区
     *          170-189 进入第 2 分区
     *          其他    进入第 3 分区
     * @param text the key to be partioned.
     * @param phoneTrafficWritable the entry value.
     * @param numPartitions the total number of partitions.
     * @return
     */
    @Override
    public int getPartition(Text text, PhoneTrafficWritable phoneTrafficWritable, int numPartitions) {

        //1.把key(手机号)转成String类型
        String phoneNum = text.toString();
        //2.截取手机号的前三位作为手机号段
        String phoneNumHeader = phoneNum.substring(0,3);
        //3.把字符串类型的号段转成int类型，方便 后面的判断
        int phoneNumHeaderInt = Integer.parseInt(phoneNumHeader);

        //4.根据不同的号段，让数据进入到不同的分区
        if(phoneNumHeaderInt>=130 && phoneNumHeaderInt<150){
            return 0;
        }else if(phoneNumHeaderInt>=150 && phoneNumHeaderInt<170){
            return 1;
        }else if(phoneNumHeaderInt>=170 && phoneNumHeaderInt<190){
            return 2;
        }

        return 3;
    }
}
